package net.bw.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author: liuyawei
 * @date: 2025/5/18
 */
public class LogSplitProcessFunction extends ProcessFunction<String, String> {

    //定义侧流数据实例
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public LogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //  1. 解析数据
        JSONObject jsonObject = JSON.parseObject(value);

        // 2. 错误日志
        if (null != jsonObject.get("err")) {
            // 输出到侧输出流
            ctx.output(errorTag, value);
            // 错误日志保存后，删除err字段
            jsonObject.remove("err");
        }

        // 3. 启动日志
        if (null != jsonObject.get("start")) {
            // 输出到侧输出流
            ctx.output(startTag, jsonObject.toJSONString());
        }
        // 4. 页面日志
        else {
            // 获取曝光日志已经动作日志
            JSONArray displaysJSONArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJSONArray = jsonObject.getJSONArray("actions");

            // 删除曝光日志和动作日志
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            // 5. 输出页面日志
            out.collect(jsonObject.toJSONString());

            // 6. 输出曝光日志和动作日志
            if (null != displaysJSONArray && !displaysJSONArray.isEmpty()) {

                // 循环遍历曝光日志JSON数组
                for (int i = 0; i < displaysJSONArray.size(); i++) {
                    // 获取曝光日志
                    JSONObject displayJSONObject = displaysJSONArray.getJSONObject(i);
                    // 设置曝光日志
                    jsonObject.put("display", displayJSONObject);
                    // 输出
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }
            // 存储曝光日志后，删除曝光日志
            jsonObject.remove("display");

            // 7. 输出动作日志
            if (null != actionsJSONArray && !actionsJSONArray.isEmpty()) {
                // 循环遍历动作日志JSON数组
                for (int i = 0; i < actionsJSONArray.size(); i++) {
                    // 获取动作日志
                    JSONObject actionJSONObject = actionsJSONArray.getJSONObject(i);
                    // 设置动作日志
                    jsonObject.put("action", actionJSONObject);
                    // 输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }

}

